package topic

import (
	"go.uber.org/atomic"
	"sync"
	"wths/wthsjy-im/plugins/log"
	"wths/wthsjy-im/protocol"
	"wths/wthsjy-im/server/handle"
)

type topic struct {
	topicName string
	clientNum atomic.Uint32
	bucket    sync.Map
}

// 保存
func (t *topic) add(cli handle.ClientHandle) uint32 {
	_, exist := t.bucket.Load(cli)
	if exist {
		log.Debug("#####add").Msg("add exist", cli.ClientID())
		return t.clientNum.Load()
	}
	log.Debug("#####add").Msg("add new", cli.ClientID())

	// 新增
	t.bucket.Store(cli, struct{}{})
	return t.clientNum.Inc()
}

// 移除
func (t *topic) remove(cli handle.ClientHandle) uint32 {
	_, exist := t.bucket.Load(cli)
	if !exist {
		log.Debug("#####remove").Msg("remove", cli.ClientID())
		return t.clientNum.Load()
	}
	t.bucket.Delete(cli)
	log.Debug("#####remove").Msg("remove 存在", cli.ClientID())

	return t.clientNum.Dec()
}

// 推送消息
func (t *topic) pushMessage(data []byte) {
	p, _ := protocol.NewPublishPacket(t.topicName, data)
	bs, err := protocol.Encode(p)
	if err != nil {
		return
	}
	log.Debug("###pushMessage").Msg(t.topicName, t.clientNum)

	t.bucket.Range(func(key, value any) bool {
		c := (key).(handle.ClientHandle)
		if !c.Closed() {
			_ = c.Write(bs)
		}
		return true
	})

}
